#include "EventLoop.h"
#include "Logger.h"
#include "Poller.h"
#include "Channel.h"

#include <sys/eventfd.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <memory>

//防止一个线程创建多个EventLoop  thread_local
__thread EventLoop* t_loopInThisThread = nullptr;

//定义默认的Poller IO复用接口的超时时间
const int kPollTimeMs = 10000;

//创建wakeupfd,用来notify唤醒subReactor处理新来的channel
/**
 * eventfd 是一个 Linux 提供的系统调用，用于创建一个特殊的文件描述符，其内部维护一个 64 位的计数器。
这个计数器的初始值通常为0。
eventfd 支持两种主要操作：写入和读取。
写入操作通过向 eventfd 写入一个整数值来增加计数器的值，读取操作则通过读取 eventfd 来获得计数器的当前值。
当计数器的值非零时，读取操作会立即返回，如果计数器的值为零，则读取操作会阻塞，直到计数器的值变为非零或者设置了非阻塞标志。
这样就可以实现生产者消费者模型
*/
int createEventfd()
{
    int evtfd = ::eventfd(0,EFD_NONBLOCK | EFD_CLOEXEC);
    if(evtfd < 0)
    {
        LOG_FATAL("eventfd error: %d",errno);
    }
    return evtfd;
}

EventLoop::EventLoop()
    : looping_(false)
    , quit_(false)
    , callingPendingFunctors_(false)
    , threadId_(CurrentThread::tid())
    , poller_(Poller::newDefaultPoller(this))
    , wakeupFd_(createEventfd())
    , wakeupChannel_(new Channel(this,wakeupFd_))
{
    LOG_DEBUG("EventLoop created %p in thread %d", this, threadId_);
    //线程已经有loop了
    if(t_loopInThisThread)
    {
        LOG_FATAL("Another EventLoop %p exists in this thread %d \n", t_loopInThisThread, threadId_);
    }
    else
    {
        t_loopInThisThread = this;
    }

    //设置wakeupfd的事件以及已经发生事件后的回调操作
    //bind绑定类成员函数时，第一个参数表示对象的成员函数的指针，第二个参数表示对象的地址，
    //这是因为对象的成员函数需要有this指针。并且编译器不会将对象的成员函数隐式转换成函数指针，需要通过&手动转换；
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead,this));
    //每一个EventLoop都将监听wakeup Channel的EPOLLIN读事件了
    wakeupChannel_->enableReading();
}

EventLoop::~EventLoop()
{
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    ::close(wakeupFd_);
    t_loopInThisThread = nullptr;
}

//开启事件循环
void EventLoop::loop()
{
    looping_ = true;
    quit_ = false;

    LOG_INFO("EventLoop %p start looping", this);

    while(!quit_)
    {
        activeChannels_.clear();
        //监听两类fd,一种是cient的fd,一种是wakeup fd
        pollReturnTime_ = poller_->poll(kPollTimeMs,&activeChannels_);
        for(Channel* channel : activeChannels_)
        {
            //poller监听哪些channel发生事件了，然后上报给EventLoop,通知channel处理相应事件
            channel->handleEvent(pollReturnTime_);
        }
        //执行当前EventLoop事件循环需要处理的回调操作
        /**
         * IO线程 (mainloop) accept fd <= channel subloop
         * mainloop事先注册一个回调cb(需要subloop阿来执行) wakeup subloop后，执行下面的方法，执行之前mainloop注册的cb
        */
        doPendingFunctors();
    }

    LOG_INFO("EventLoop %p stop looping \n",this);
    looping_ = false;

}

//退出事件循环   1.loop在自己的线程中调用quit  2.在非loop的线程中，调用loop的quit
/**
 *              mainLoop
 * 
 *                                    no ==================生产者-消费者的线程安全的队列
 * 
 * subLoop1    subLoop2     subloop3     
*/
void EventLoop::quit()
{
    quit_ = true;

    //如果是在其它线程中调用了quit,比如在subloop中调用了mainloop的quit，那么需要先把mainloop线程唤醒，不然执行不到quit_
    if(!isInLoopThread())
    {
        wakeup();
    }
}

//在当前loop中执行cb
void EventLoop::runInLoop(Functor cb)
{
    if(isInLoopThread()) //在当前loop线程中执行cb
    {
        cb();
    }
    else //在非当前loop线程中执行cb,就需要唤醒loop所在线程执行cb
    {
        queueInLoop(cb);
    }
}

//把cb放入队列(vector)中，唤醒loop所在线程，执行cb
void EventLoop::queueInLoop(Functor cb)
{
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFunctors_.emplace_back(cb);
    }

    //唤醒相应的，需要执行上面回调操作的loop线程
    /**
     * 判断callingPendingFunctors_是考虑到了要唤醒的loop线程正在执行回调的情况，但是loop又有了新的回调，
     * 这个时候唤醒的话等到loop执行玩回调再次阻塞在poll的时候又会被唤醒
    */
    if(!isInLoopThread() || callingPendingFunctors_) 
    {
        wakeup(); //唤醒loop所在线程
    }
}

//读取wakeup Channel上的消息
void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one , sizeof(one));
    if(n!=sizeof(one))
    {
        LOG_ERROR("EventLoop::handleRead() reads %lu bytes instead of 8",n);
    }
}

//唤醒subloop(subReactor) 所在线程  向wakeupfd_写一个数据即可,wakeupChannel就发生读事件，当前loop线程就会被唤醒
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof(one));
    if(n != sizeof(one))
    {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8",n);
    }
}

//EventLoop的方法=>Poller的方法
void EventLoop::updateChannel(Channel* channel)
{
    poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel* channel)
{
    poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel* channel)
{
    return poller_->hasChannel(channel);
}

//执行回调
void EventLoop::doPendingFunctors()
{
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;

    {
        std::unique_lock<std::mutex> lock(mutex_);
        //这样可以快速解放pendingFunctors_，不会影响到mainloop访问pendingFunctors_
        functors.swap(pendingFunctors_);
    }

    for(const Functor &functor : functors)
    {
        functor(); //执行当前loop需要执行的回调操作
    }

    callingPendingFunctors_ = false;
}

